package com.shujia.sql

import org.apache.spark.SparkContext
import org.apache.spark.sql.{DataFrame, SQLContext, SparkSession}

object Demo01WordCountOnSparkSQL {
  def main(args: Array[String]): Unit = {
    /**
     * 构建SparkSession环境，Spark从2.x系列引入的统一的入口
     */
    val spark: SparkSession = SparkSession
      .builder()
      .appName("Demo01WordCountOnSparkSQL")
//      .master("local")
      .config("spark.default.parallelism", "4")
      .getOrCreate()

    // 可以直接通过SparkSession获取SparkContext
    val sc: SparkContext = spark.sparkContext

    // 读数据
    val wordsDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", "#")
      .schema("words String")
      .load("/spark/data/wc/")

    // 打印数据：show
    wordsDF.show() // 默认打印20条数据
    wordsDF.show(5) // 指定输出的数据条数
    wordsDF.show(truncate = false) // 展示完整的数据

    // 使用RDD的方式处理
    wordsDF
      .rdd
      .map(row => {
        val words: String = row.getAs[String]("words")
        words
      })
      .flatMap(_.split(","))
      .map(word => (word, 1))
      .reduceByKey(_ + _)
      .foreach(println)

    // Spark SQL的方式

    // 第一种方式：SQL
    // 将DF注册成一张表
    wordsDF.createOrReplaceTempView("words_tb")
    // 写SQL 最后不需要加分号结尾，加了反而会报错
    val wordCntDF: DataFrame = spark.sql(
      """
        |select  t1.word
        |        ,count(*) as cnt
        |from (
        |    select  explode(split(words,",")) as word
        |    from words_tb
        |) t1 group by t1.word
        |""".stripMargin)
    wordCntDF.show() // 需要“Action算子”触发

    // 第二种方式：DSL
    /**
     * 字符串表达式
     * 列表达式
     */
    /**
     * 导入隐式转换以及Spark SQL提供的函数包
     */
    import spark.implicits._
    import org.apache.spark.sql.functions._

    wordsDF
      .select(explode(split($"words", ",")) as "word")
      .groupBy($"word")
      .agg(count("*") as "cnt")
      .show()


  }

  /**
   * 提交Spark-SQL代码到集群：
   * 1、注释master
   * 2、将本地路径修改成HDFS的路径
   * 3、准备数据
   * 4、打包上传提交任务：
   *  spark-submit --master yarn-client --class com.shujia.sql.Demo01WordCountOnSparkSQL spark-1.0.jar
   */

}
